package com.dtwave.cfstreaming.utils;

import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class HbaseUtils {
    public static Logger LOGGER = LoggerFactory.getLogger(HbaseUtils.class);

    /**
     * 依据rowkey和列簇将数据写入hbase
     */
    public static void put(Connection connection, String tableName, String rowKey,String familyName,String key,String value) throws Exception{
        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(key),Bytes.toBytes(value));
            table.put(put);
        }catch (Exception e){
            LOGGER.error("数据插入失败{1}",e);
        }finally {
            extracted(table);
        }
    }

    /**
     * 依据rowkey和列簇将json写入hbase
     */
    public static void put(Connection connection, String tableName, String rowKey,String familyName,JSONObject json) throws Exception {
        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowKey));
            for (String s : json.keySet()) {
                put.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(s),Bytes.toBytes(json.get(s)+""));
            }
            table.put(put);
        }catch (Exception e){
            LOGGER.error("数据插入失败{1}",e);
        }finally {
            extracted(table);
        }
    }



        /**
         * 依据rowkey查询
         * @param connection
         * @param tableName
         * @param rowKey
         * @return
         * @throws Exception
         */
    public static JSONObject getJSONObjectFromHbase(Connection connection, String tableName, String rowKey) throws Exception {
        LOGGER.debug("依据表名获取hbase数据");
        Table table = null;
        String value;
        String key;
        JSONObject jObj = new JSONObject();
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            for (Cell cell  : result.rawCells()) {
                key = Bytes.toString(CellUtil.cloneQualifier(cell));
                value = new String(CellUtil.cloneValue(cell));
                jObj.put(key,value);
            }
        }catch (Exception e){
            LOGGER.error("hbase获取数据发生异常");
        }finally {
            extracted(table);
        }
        return jObj;
    }

    /**
     * 关闭table
     * @param table
     */
    private static void extracted(Table table) {
        try {
            if(table != null){
                table.close();
            }
        }catch (Exception e){
            LOGGER.error("hbase关闭表连接失败");
        }
    }
}
